package com.stone.zeus.component.grpc.integration.config;

import com.google.common.collect.Maps;
import com.zb.zeus.common.core.domain.ZeusProjectProperty;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import java.util.HashMap;

/**
 * @author fujianjian
 * @date 2023/3/21 17:03
 **/
@Configuration
@RequiredArgsConstructor
@ConditionalOnProperty(prefix = "zeus.project.perf", value = "enable-recv", havingValue = "true")
public class KafkaConfiguration {

    private final ZeusProjectProperty zeusProjectProperty;

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        HashMap<String, Object> configMap = Maps.newHashMap();
        configMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, zeusProjectProperty.getPerf().getKafkaBroker());
        configMap.put(ConsumerConfig.GROUP_ID_CONFIG, zeusProjectProperty.getPerf().getRecvGroupId());
        configMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        configMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        ConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(configMap);

        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConcurrency(2);
        factory.setConsumerFactory(consumerFactory);
        return factory;
    }
}
